package com.example.mqttdemo.service;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

@Service
@Slf4j
public class MqttService implements MqttCallback {

    @Value("${mqtt.url}")
    private String url;

    @Value("${mqtt.username}")
    private String username;

    @Value("${mqtt.password}")
    private String password;

    @Value("${mqtt.clientId}")
    private String clientId;

    @Value("${mqtt.cleanSession}")
    private boolean cleanSession;

    private MqttClient client;

    private int reconnectDelay = 2000; // 初始重连延迟2秒
    private int maxReconnectAttempts = 3; // 最大重连尝试次数
    private int reconnectAttempts = 0;

    @PostConstruct
    public void connect() {
        try {
            client = new MqttClient(url, clientId);
            MqttConnectOptions options = new MqttConnectOptions();
            // cleanSession为 false 时表示创建一个持久会话，在客户端断开连接时，会话仍然保持并保存离线消息，直到会话超时注销。
            // cleanSession为 true 时表示创建一个新的临时会话，在客户端断开时，会话自动销毁。
            // 注意：持久会话恢复的前提是客户端使用固定的 Client ID 再次连接，如果 Client ID 是动态的，那么连接成功后将会创建一个新的持久会话。
            options.setCleanSession(cleanSession);
            // 禁用Paho的自动重连，自己控制
            options.setAutomaticReconnect(false);
            if (username != null && !username.isEmpty()) {
                options.setUserName(username);
                options.setPassword(password.toCharArray());
            }
            client.setCallback(this);
            client.connect(options);

            // 订阅一个或多个主题
            client.subscribe("test/topic");
        } catch (MqttException e){
            log.error("---mqtt connect fail", e);
        }

    }

    // 实现MqttCallback的方法：connectionLost, messageArrived, deliveryComplete

    @Override
    public void connectionLost(Throwable cause) {
        log.info("---Connection lost! " + cause.getMessage());
        // 这里可以重新连接MQTT服务器
        reconnect();
    }

    private void reconnect() {
        if (reconnectAttempts < maxReconnectAttempts) {
            reconnectAttempts++;
            log.info("---Attempting to reconnect in " + reconnectDelay + "ms");

            // 使用ScheduledExecutorService方式实现延迟重连
            // 为简单起见，使用Thread.sleep
            try {
                Thread.sleep(reconnectDelay);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }

            reconnectDelay *= 2; // 增大重连间隔
            connect(); // 尝试重新连接
        } else {
            log.warn("---Max reconnect attempts reached");
            // 可以考虑执行一些清理操作或通知操作
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        // 当消息到达时调用
        log.info("---Message arrived. Topic: " + topic + " Message: " + new String(message.getPayload()));
        // 处理消息的逻辑
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // 当消息被完全传送出去后调用
        log.info("---Delivery complete!");
        // 可以在这里处理一些发送完成后的清理工作
    }

    // 发送消息的方法
    public void publish(String topic, String payload) throws MqttException {
        MqttMessage message = new MqttMessage(payload.getBytes());
        // QoS 0，最多交付一次。可能丢失消息
        // QoS 1，至少交付一次。可以保证收到消息，但消息可能重复
        // QoS 2，只交付一次。可以保证消息既不丢失也不重复
        message.setQos(2);
        client.publish(topic, message);
        log.info("---Message published： {}", payload);
    }

    // 断开连接的方法
    @PreDestroy
    public void disconnect() throws MqttException {
        if (client != null && client.isConnected()) {
            client.disconnect();
            log.info("---Disconnected");
        }
    }
}
